package com.supplychain.scm.core.rocketmq.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.supplychain.scm.core.basis.constant.CommonConstants;
import com.supplychain.scm.core.basis.constant.StringPool;
import com.supplychain.scm.core.rocketmq.bean.BaseMessageConsumer;
import com.supplychain.scm.core.rocketmq.bean.MessageTypeEnum;
import com.supplychain.scm.core.rocketmq.constant.RocketMqConstant;
import com.supplychain.scm.core.rocketmq.exception.RocketMqBizException;
import com.supplychain.scm.core.rocketmq.handler.AbstractBroadcastMessageHandler;
import com.supplychain.scm.core.rocketmq.storage.IMqMessageConsumerStorage;
import com.supplychain.scm.core.rocketmq.util.RocketThreadLocalUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.Executor;
import java.util.stream.Collectors;

/**
 * @author liyang
 * @version 1.0.0
 * @ClassName AbstractBroadcastListener
 * @Description 抽象广播消息监听器
 * @createTime 10:47:00 2022/12/05
 */
@Slf4j
@RefreshScope
public abstract class AbstractBroadcastListener implements RocketMQListener<MessageExt> {

    @Resource
    protected List<AbstractBroadcastMessageHandler> broadcastMessageHandlers;
    @Autowired(required = false)
    @Qualifier(value = RocketMqConstant.EXECUTOR_NAME)
    protected Executor executor;
    @Autowired(required = false)
    protected IMqMessageConsumerStorage mqMessageConsumerStorage;
    @Value("${spring.application.name}")
    protected String applicationName;
    @Value("${rocketmq.consumer.store.enabled:true}")
    protected Boolean ifStore;
    protected int orderlyMaxRetryTimes = 20;

    protected final void setBroadcastMessageHandlers(List<AbstractBroadcastMessageHandler> broadcastMessageHandlers){
        if(CollectionUtils.isEmpty(this.broadcastMessageHandlers) && !CollectionUtils.isEmpty(broadcastMessageHandlers)){
            this.broadcastMessageHandlers = broadcastMessageHandlers;
        }
    }

    protected void beforeHandler(MessageExt message){

    }

    protected void afterHandler(MessageExt message,boolean result){

    }

    protected void failHandler(MessageExt message){

    }
    protected void finallyHandler(MessageExt message){

    }
    protected void exceptionHandler(MessageExt message,Exception exception){

    }

    /**
     * 控制是否重试 false - 重试; true - 不重试
     * @param message
     * @return
     */
    protected boolean ifRetry(MessageExt message,Exception exception) {
        return false;
    }

    @Override
    public void onMessage(MessageExt message) {
        if(log.isDebugEnabled()) {
            log.debug("api广播消息Consumer接收到消息，messageid={}，message={},topic = {},tag = {}", message.getMsgId(), message, message.getTopic(), message.getTags());
        }
        List<AbstractBroadcastMessageHandler> handlers = broadcastMessageHandlers.stream()
                .filter(handler -> StringUtils.equals(handler.getTopic(), message.getTopic()) && StringUtils.equals(handler.getTag(), message.getTags()))
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(handlers) || handlers.size() > 1) {
            log.warn("未定义或定义多个消息处理程序，message={}", message);
            return;
        }
        boolean result = false;
        try {
            RocketThreadLocalUtils.setMessageHead(message.getProperties());
            this.beforeHandler(message);
            result = handlers.stream().findAny().orElseThrow(() -> new RocketMqBizException(String.format("toipc:[%s] tag:[%s] consumer handler not null",message.getTopic(),message.getTags()))).handler(message);
            this.afterHandler(message,result);
            if(log.isDebugEnabled()) {
                log.debug("api广播消息Consumer接收到消息，message={},处理结果 result={}", message, result);
            }
            if(!result) {
                this.failHandler(message);
            }
        } catch (Exception e) {
            log.error("message:{}, data:{}, error:{}", "广播消息Consumer异常", JSON.toJSONString(message, SerializerFeature.IgnoreErrorGetter), ExceptionUtils.getStackTrace(e));
            this.exceptionHandler(message, e);
            if(!this.ifRetry(message,e)) {
                if (this.orderlyMessageRetry(message)) {
                    throw e;
                }
            }
        }finally {
            this.finallyHandler(message);
            //目前只记录消费成功的
            //Environment environment = SpringContextHolder.getBean(Environment.class);
            //String applicationName = environment.getProperty("spring.application.name");
            BaseMessageConsumer baseMessageConsumer = this.buildBaseMessageConsumer(message,result);
            RocketThreadLocalUtils.remove();
            if(log.isDebugEnabled()) {
                log.debug("consumer 广播消息 baseMessageConsumer = {}",JSON.toJSONString(baseMessageConsumer));
            }
            if(result && Objects.nonNull(mqMessageConsumerStorage)) {
                if(!RocketMqConstant.EXCLUDE_TOPICS.contains(Optional.ofNullable(baseMessageConsumer).map(BaseMessageConsumer::getTopic).orElse(StringPool.EMPTY)) &&
                        !RocketMqConstant.EXCLUDE_TAGS.contains(Optional.ofNullable(baseMessageConsumer).map(BaseMessageConsumer::getTag).orElse(StringPool.EMPTY))) {
                    mqMessageConsumerStorage.createBaseMessageConsumer(baseMessageConsumer,executor);
                }
            }
        }
    }

    private boolean orderlyMessageRetry(MessageExt message){
        Map<String, String> messageHead = RocketThreadLocalUtils.getMessageHead();
        String messageType = Optional.ofNullable(messageHead).map((head) -> head.get(RocketMqConstant.MESSAGE_TYPE)).orElse(StringPool.EMPTY);
        if(StringUtils.isNotBlank(messageType) && Objects.equals(messageType, MessageTypeEnum.ORDERLY_MESSAGE.getValue()) &&
                (message.getReconsumeTimes() >= orderlyMaxRetryTimes) && (orderlyMaxRetryTimes > CommonConstants.ZERO)) {
            if(log.isDebugEnabled()) {
                log.debug("顺序消息 重试次数 {} 次，不在继续重试 message = {}",message.getReconsumeTimes(),JSON.toJSONString(message, SerializerFeature.IgnoreErrorGetter));
            }
            return Boolean.FALSE;
        }
        return Boolean.TRUE;
    }

    private BaseMessageConsumer buildBaseMessageConsumer(MessageExt message,boolean ifConsumer){

        return BaseMessageConsumer.builder()
                .topic(message.getTopic())
                .tag(message.getTags())
                .msgId(message.getMsgId())
                .status(ifConsumer ? RocketMqConstant.SUCCESS : RocketMqConstant.FAIL)
                .uniqueKey(RocketThreadLocalUtils.getUniqueKey())
                .messageLinkId(RocketThreadLocalUtils.getMessageLinkId())
                .messageOrigin(RocketThreadLocalUtils.getMessageOrigin())
                .origin(applicationName)
                .ifStore(ifStore)
                .build();
    }
}
